package com.yangrong.thread.chapter13;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

public class SimpleThreadPool extends Thread {

	private int size;

	private final int queueSize;

	private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;

	private static volatile int seq = 0;

	private final static String THREAD_PREFIX = "SIMPLE_THREAD_POOL-";

	private final static ThreadGroup GROUP = new ThreadGroup("Pool_Group");

	// 线程队列
	private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<Runnable>();

	private final static List<WorkerTask> THREAD_QUEUE = new ArrayList<>();

	private final DiscardPolicy discardPolicy;

	public final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
		throw new DiscardException("Discard this task.");
	};

	private volatile boolean destroy = false;

	private int min;

	private int max;

	private int active;

	public SimpleThreadPool() {
		this(4, 8, 12, DEFAULT_TASK_QUEUE_SIZE, DEFAULT_DISCARD_POLICY);
	}

	public SimpleThreadPool(int min, int active, int max, int queueSize, DiscardPolicy discardPolicy) {
		this.min = min;
		this.active = active;
		this.max = max;
		this.queueSize = queueSize;
		this.discardPolicy = discardPolicy;
		init();
	}

	private void init() {
		for (int i = 0; i < this.min; i++) {
			createWorkTask();
		}
		this.size = min;
		this.start();
	}

	public void submit(Runnable runnable) {
		if (destroy)
			throw new IllegalStateException("The thread pool already distroy and not allow submit task.");

		synchronized (TASK_QUEUE) {
			if (TASK_QUEUE.size() > queueSize)
				discardPolicy.discard();
			TASK_QUEUE.addLast(runnable);
			TASK_QUEUE.notifyAll();
		}
	}

	private void createWorkTask() {
		WorkerTask task = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
		task.start();
		THREAD_QUEUE.add(task);
	}

	@Override
	public void run() {
		while (!destroy) {
			System.out.printf("Pool#Min:%d,Active:%d,Max:%d,Current:%d,QueueSize:%d\n", this.min, this.active, this.max,
					this.size, TASK_QUEUE.size());
			try {
				Thread.sleep(5_000);
				if (TASK_QUEUE.size() > active && size < active) {
					for (int i = size; i < active; i++) {
						createWorkTask();
					}
					System.out.println("The pool incremented to active.");
					size = active;
				} else if (TASK_QUEUE.size() > max && size < max) {
					for (int i = size; i < max; i++) {
						createWorkTask();
					}
					System.out.println("The pool incremented to max.");
					size = max;
				}
				synchronized (THREAD_QUEUE) {
					if (TASK_QUEUE.isEmpty() && size > active) {
						System.out.println("======Reduce======");
						int releaseSize = size - active;
						for (Iterator<WorkerTask> it = THREAD_QUEUE.iterator(); it.hasNext();) {
							if (releaseSize <= 0)
								break;
							WorkerTask task = it.next();
							task.close();
							task.interrupt();
							it.remove();
							releaseSize--;
						}
						size = active;
					}
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public void shutdown() throws InterruptedException {
		while (!TASK_QUEUE.isEmpty()) {
			Thread.sleep(50);
		}

		synchronized (THREAD_QUEUE) {
			int initVal = THREAD_QUEUE.size();
			while (initVal > 0) {
				for (WorkerTask task : THREAD_QUEUE) {
					if (task.getTaskState() == TaskState.BLICKED) {
						task.interrupt();
						task.close();
						initVal--;
					} else {
						Thread.sleep(10);
					}
				}
			}
		}

		this.destroy = true;
		System.out.println("The thread pool disposed.");
	}

	public int getQueueSize() {
		return queueSize;
	}

	public int getSize() {
		return size;
	}

	public boolean isDestroy() {
		return this.destroy;
	}

	public int getMin() {
		return min;
	}

	public int getMax() {
		return max;
	}

	public int getActive() {
		return active;
	}

	private enum TaskState {
		FREE, RUNNING, BLICKED, DEAD
	}

	public static class DiscardException extends RuntimeException {

		public DiscardException(String message) {
			super(message);
		}
	}

	public interface DiscardPolicy {

		void discard() throws DiscardException;
	}

	private static class WorkerTask extends Thread {

		private volatile TaskState taskState = TaskState.FREE;

		public WorkerTask(ThreadGroup group, String name) {
			super(group, name);
		}

		public TaskState getTaskState() {
			return this.taskState;
		}

		@Override
		public void run() {
			OUTER: while (this.taskState != TaskState.DEAD) {
				Runnable runnable;
				synchronized (TASK_QUEUE) {
					while (TASK_QUEUE.isEmpty()) {
						try {
							taskState = TaskState.BLICKED;
							TASK_QUEUE.wait();
						} catch (InterruptedException e) {
							break OUTER;
						}
					}

					runnable = TASK_QUEUE.removeFirst();

					// 错误示例，线程执行不能在synchronized代码块中，这样会使线程池在线程执行结束前都处在lock状态
//					taskState = TaskState.RUNNING;
//					runnable.run();
//					taskState = TaskState.FREE;
				}
				if (runnable != null) {
					taskState = TaskState.RUNNING;
					runnable.run();
					taskState = TaskState.FREE;
				}
			}
		}

		public void close() {
			this.taskState = TaskState.DEAD;
		}
	}

	public static void main(String[] args) throws InterruptedException {
//		SimpleThreadPool simpleThreadPool = new SimpleThreadPool(6, 10, SimpleThreadPool.DEFAULT_DISCARD_POLICY);
		SimpleThreadPool simpleThreadPool = new SimpleThreadPool();
		for (int i = 0; i < 40; i++) {
			simpleThreadPool.submit(() -> {
				System.out.println("The runnable be serviced by " + Thread.currentThread() + " Start.");
				try {
					Thread.sleep(3000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				System.out.println("The runnable be serviced by " + Thread.currentThread() + " finished.");
			});
		}

		Thread.sleep(10_000);
		simpleThreadPool.shutdown();
//		try {
//			Thread.sleep(10000);
//			simpleThreadPool.shutdown();
//			simpleThreadPool.submit(() -> System.out.println("shutdown后再取线程"));
//		} catch (InterruptedException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
	}

}
